//! # OpenCloudOS Node Agent
//!
//! The agent component that runs on managed nodes. It connects to the central
//! server and executes commands, reports results, and maintains heartbeat.
//!
//! ## Features
//! - Automatic registration with server
//! - Group-based command routing
//! - Heartbeat/health monitoring
//! - Script execution with timeout support
//! - File transfer capabilities
//! - TLS/SSL encrypted communication
//! - Token-based authentication
//!
//! ## Usage
//!
//! Basic usage:
//! ```bash
//! node_agent [--groups GROUP1,GROUP2] [--enable-tls]
//! ```
//!
//! With custom configuration:
//! ```bash
//! node_agent --groups web,database --token mytoken --enable-tls
//! ```

use anyhow::Result;
use base64::{engine::general_purpose, Engine as _};
use clap::Parser;
use lapin::{
    options::*, types::FieldTable, BasicProperties, Channel, Connection, ConnectionProperties,
};
use shared::protocol::{queues, AgentRegistration, HeartbeatMessage, ResultMessage, TaskMessage};
use std::process::Command;
use tokio::time::Duration;
use uuid::Uuid;

/// Command-line interface for the node agent
#[derive(Parser)]
#[command(name = "node_agent")]
#[command(about = "OpenCloudOS 节点代理")]
struct Cli {
    #[arg(long, default_value = "amqp://admin:admin123@127.0.0.1:5672")]
    rabbitmq_url: String,

    #[arg(long, default_value = "web,default")]
    groups: String,

    #[arg(long, default_value = "secure_token_123")]
    token: String,

    #[arg(long)]
    enable_tls: bool,

    #[arg(long, default_value = "./certs/ca.crt")]
    ca_path: String,
}

/// Main agent structure handling all node operations
///
/// Manages connection to the server, executes received commands,
/// and maintains heartbeat for health monitoring.
struct NodeAgent {
    /// Unique identifier for this agent instance
    agent_id: String,
    /// System hostname
    hostname: String,
    /// List of groups this agent belongs to
    groups: Vec<String>,
    /// Authentication token for server communication
    token: String,
    /// RabbitMQ channel for message operations
    rabbitmq_channel: Channel,
    /// TLS configuration for secure connections
    tls_config: Option<shared::tls::TlsConfig>,
}

impl NodeAgent {
    /// Create a new NodeAgent instance
    ///
    /// Establishes connection to RabbitMQ and sets up the agent with
    /// the provided configuration.
    ///
    /// # Arguments
    /// * `rabbitmq_url` - RabbitMQ connection URL
    /// * `groups` - List of groups this agent belongs to
    /// * `token` - Authentication token
    /// * `tls_config` - Optional TLS configuration
    async fn new(
        rabbitmq_url: &str,
        groups: Vec<String>,
        token: String,
        tls_config: Option<shared::tls::TlsConfig>,
    ) -> Result<Self> {
        let agent_id = Uuid::new_v4().to_string();
        let hostname = Self::get_hostname();

        // Validate TLS configuration
        if let Some(ref config) = tls_config {
            shared::tls::validate_cert_files(config)?;
            println!("🔒 TLS enabled for RabbitMQ connections");
        }

        // Connect to RabbitMQ
        let conn = Connection::connect(rabbitmq_url, ConnectionProperties::default()).await?;
        let rabbitmq_channel = conn.create_channel().await?;

        Ok(Self {
            agent_id,
            hostname,
            groups,
            token,
            rabbitmq_channel,
            tls_config,
        })
    }

    /// Get the hostname of the current system
    ///
    /// # Returns
    /// Returns the system hostname, or "unknown" if hostname cannot be determined.
    ///
    /// # Examples
    /// ```
    /// let hostname = NodeAgent::get_hostname();
    /// println!("Host: {}", hostname);
    /// ```
    fn get_hostname() -> String {
        Command::new("hostname")
            .output()
            .map(|output| String::from_utf8_lossy(&output.stdout).trim().to_string())
            .unwrap_or_else(|_| {
                eprintln!("⚠️ Unable to get hostname, using 'unknown'");
                "unknown".to_string()
            })
    }

    /// Get operating system information from /etc/os-release
    ///
    /// # Returns
    /// Returns the pretty name of the OS if available, otherwise "Unknown OS".
    ///
    /// # Notes
    /// This function specifically looks for the PRETTY_NAME field in /etc/os-release,
    /// which is the standard location for OS information on most Linux distributions.
    fn get_os_info() -> String {
        match std::fs::read_to_string("/etc/os-release") {
            Ok(content) => {
                for line in content.lines() {
                    if line.starts_with("PRETTY_NAME=") {
                        return line
                            .strip_prefix("PRETTY_NAME=")
                            .unwrap_or("unknown")
                            .trim_matches('"')
                            .to_string();
                    }
                }
                "Unknown OS".to_string()
            }
            Err(e) => {
                eprintln!("⚠️ Unable to read /etc/os-release: {}", e);
                "Unknown OS".to_string()
            }
        }
    }

    async fn start(&self) -> Result<()> {
        // Setup RabbitMQ
        self.setup_rabbitmq().await?;

        // Register agent
        self.register().await?;

        // Start heartbeat task
        let heartbeat_agent = self.clone_for_heartbeat();
        tokio::spawn(async move {
            heartbeat_agent.heartbeat_task().await;
        });

        // Start task consumer
        self.start_task_consumer().await?;

        Ok(())
    }

    fn clone_for_heartbeat(&self) -> Self {
        Self {
            agent_id: self.agent_id.clone(),
            hostname: self.hostname.clone(),
            groups: self.groups.clone(),
            token: self.token.clone(),
            rabbitmq_channel: self.rabbitmq_channel.clone(),
            tls_config: self.tls_config.clone(),
        }
    }

    /// Setup RabbitMQ exchanges, queues, and bindings for this agent
    ///
    /// This function:
    /// 1. Declares required exchanges (task and result exchanges)
    /// 2. Creates an agent-specific task queue
    /// 3. Binds the agent queue to the task exchange for each group this agent belongs to
    ///
    /// # Returns
    /// `Result<()>` - Ok if setup is successful, Err with details if any step fails
    ///
    /// # Errors
    /// May return errors for:
    /// - Exchange declaration failures
    /// - Queue creation failures  
    /// - Queue binding failures
    async fn setup_rabbitmq(&self) -> Result<()> {
        // Declare task exchange for routing commands to agents
        if let Err(e) = self
            .rabbitmq_channel
            .exchange_declare(
                queues::TASK_EXCHANGE,
                lapin::ExchangeKind::Topic,
                ExchangeDeclareOptions::default(),
                FieldTable::default(),
            )
            .await
        {
            return Err(anyhow::anyhow!("Failed to declare task exchange: {}", e));
        }

        // Declare result exchange for routing results back to server
        if let Err(e) = self
            .rabbitmq_channel
            .exchange_declare(
                queues::RESULT_EXCHANGE,
                lapin::ExchangeKind::Direct,
                ExchangeDeclareOptions::default(),
                FieldTable::default(),
            )
            .await
        {
            return Err(anyhow::anyhow!("Failed to declare result exchange: {}", e));
        }

        // Create agent-specific task queue
        let agent_queue_name = format!("agent_{}", self.agent_id);
        if let Err(e) = self
            .rabbitmq_channel
            .queue_declare(
                &agent_queue_name,
                QueueDeclareOptions::default(),
                FieldTable::default(),
            )
            .await
        {
            return Err(anyhow::anyhow!(
                "Failed to create agent queue '{}': {}",
                agent_queue_name,
                e
            ));
        }

        // Bind agent queue to task exchange for each group
        for group in &self.groups {
            if let Err(e) = self
                .rabbitmq_channel
                .queue_bind(
                    &agent_queue_name,
                    queues::TASK_EXCHANGE,
                    group,
                    QueueBindOptions::default(),
                    FieldTable::default(),
                )
                .await
            {
                return Err(anyhow::anyhow!(
                    "Failed to bind queue to group '{}': {}",
                    group,
                    e
                ));
            }
        }

        println!("✅ RabbitMQ setup completed for agent: {}", self.agent_id);
        Ok(())
    }

    /// Register this agent with the server
    ///
    /// Sends an agent registration message containing:
    /// - Agent ID and hostname
    /// - Groups this agent belongs to
    /// - Authentication token
    /// - Operating system information
    ///
    /// # Returns
    /// `Result<()>` - Ok if registration is successful, Err if message publishing fails
    ///
    /// # Errors
    /// May return errors for:
    /// - JSON serialization failures
    /// - Message publishing failures to the registration queue
    async fn register(&self) -> Result<()> {
        let registration = AgentRegistration::new(
            self.agent_id.clone(),
            self.hostname.clone(),
            self.groups.clone(),
            self.token.clone(),
            Self::get_os_info(),
        );

        let payload = match serde_json::to_vec(&registration) {
            Ok(payload) => payload,
            Err(e) => return Err(anyhow::anyhow!("Failed to serialize registration: {}", e)),
        };

        if let Err(e) = self
            .rabbitmq_channel
            .basic_publish(
                "",
                queues::REGISTRATION_QUEUE,
                BasicPublishOptions::default(),
                &payload,
                BasicProperties::default(),
            )
            .await
        {
            return Err(anyhow::anyhow!(
                "Failed to publish registration message: {}",
                e
            ));
        }

        println!("📝 Agent {} registered successfully", self.agent_id);
        Ok(())
    }

    /// Continuously send heartbeat messages to the server
    ///
    /// This function runs in a loop, sending heartbeat messages every 60 seconds
    /// to inform the server that this agent is still alive and operational.
    ///
    /// # Behavior
    /// - Sends heartbeat every 60 seconds
    /// - Continues running even if individual heartbeat messages fail
    /// - Logs errors but doesn't terminate the heartbeat loop
    /// - This function never returns unless the program is terminated
    ///
    /// # Error Handling
    /// Individual errors are logged but don't stop the heartbeat process:
    /// - Serialization errors: logged and skipped
    /// - Publishing errors: logged but heartbeat continues
    async fn heartbeat_task(&self) {
        let mut interval = tokio::time::interval(Duration::from_secs(60));
        println!("💓 Starting heartbeat task (interval: 60s)");

        loop {
            interval.tick().await;

            let heartbeat = HeartbeatMessage::new(self.agent_id.clone());
            let payload = match serde_json::to_vec(&heartbeat) {
                Ok(p) => p,
                Err(e) => {
                    eprintln!("❌ Failed to serialize heartbeat: {}", e);
                    continue;
                }
            };

            match self
                .rabbitmq_channel
                .basic_publish(
                    "",
                    queues::HEARTBEAT_QUEUE,
                    BasicPublishOptions::default(),
                    &payload,
                    BasicProperties::default(),
                )
                .await
            {
                Ok(_) => println!("💓 Heartbeat sent successfully"),
                Err(e) => eprintln!("❌ Failed to send heartbeat: {}", e),
            }
        }
    }

    /// Start consuming and processing tasks from the agent's task queue
    ///
    /// This function establishes a consumer for the agent's dedicated queue and
    /// processes incoming tasks in an infinite loop. Each task is:
    /// 1. Deserialized from JSON
    /// 2. Executed based on its command type
    /// 3. Result sent back to server
    /// 4. Message acknowledged to RabbitMQ
    ///
    /// # Returns
    /// `Result<()>` - Returns Ok on successful consumer setup, Err if consumer creation fails.
    /// Note: This function runs indefinitely once started successfully.
    ///
    /// # Error Handling
    /// - Consumer setup errors: returned immediately
    /// - Individual message processing errors: logged but don't stop the consumer
    /// - Task execution errors: captured and sent as error results to server
    async fn start_task_consumer(&self) -> Result<()> {
        let agent_queue_name = format!("agent_{}", self.agent_id);

        let consumer = match self
            .rabbitmq_channel
            .basic_consume(
                &agent_queue_name,
                "task_consumer",
                BasicConsumeOptions::default(),
                FieldTable::default(),
            )
            .await
        {
            Ok(consumer) => consumer,
            Err(e) => return Err(anyhow::anyhow!("Failed to create task consumer: {}", e)),
        };

        println!(
            "🎯 Agent {} listening for tasks on queue '{}'...",
            self.agent_id, agent_queue_name
        );

        use futures_lite::stream::StreamExt;
        let mut consumer_stream = consumer;
        while let Some(delivery) = consumer_stream.next().await {
            match delivery {
                Ok(delivery) => {
                    // Try to deserialize the task message
                    match serde_json::from_slice::<TaskMessage>(&delivery.data) {
                        Ok(task) => {
                            println!(
                                "📨 Received task: {} ({})",
                                task.command_id, task.command_type
                            );

                            // Execute the task and send result
                            let result = self.execute_task(task).await;
                            if let Err(e) = self.send_result(result).await {
                                eprintln!("❌ Failed to send task result: {}", e);
                            }
                        }
                        Err(e) => {
                            eprintln!("❌ Failed to deserialize task message: {}", e);
                        }
                    }

                    // Always acknowledge the message to prevent redelivery
                    if let Err(e) = delivery.ack(BasicAckOptions::default()).await {
                        eprintln!("❌ Failed to acknowledge message: {}", e);
                    }
                }
                Err(e) => {
                    eprintln!("❌ Failed to receive message from queue: {}", e);
                }
            }
        }

        // This should never be reached in normal operation
        Ok(())
    }

    /// Execute a received task based on its command type
    ///
    /// This function dispatches tasks to appropriate handlers based on the command type:
    /// - `get_hostname`: Returns the system hostname
    /// - `run_script`: Executes a shell script with timeout support
    /// - `receive_file`: Writes file content to local filesystem
    ///
    /// # Arguments
    /// * `task` - The task message containing command details and payload
    ///
    /// # Returns
    /// `ResultMessage` - Always returns a result (either success or error)
    /// containing execution status, output, and metadata
    ///
    /// # Supported Commands
    /// - **get_hostname**: Simple hostname retrieval (no additional parameters)
    /// - **run_script**: Script execution with Base64-encoded script content
    /// - **receive_file**: File creation with Base64-encoded file content and path
    async fn execute_task(&self, task: TaskMessage) -> ResultMessage {
        println!(
            "🔧 Executing task '{}' of type '{}'",
            task.command_id, task.command_type
        );

        match task.command_type.as_str() {
            "get_hostname" => {
                let hostname = Self::get_hostname();
                ResultMessage::success(
                    task.command_id,
                    self.agent_id.clone(),
                    serde_json::json!({
                        "hostname": hostname
                    }),
                )
            }
            "run_script" => self.execute_run_script_task(task).await,
            "receive_file" => self.execute_receive_file_task(task).await,
            _ => {
                eprintln!("❌ Unsupported command type: {}", task.command_type);
                ResultMessage::error(
                    task.command_id,
                    self.agent_id.clone(),
                    format!("Unsupported command type: '{}'", task.command_type),
                )
            }
        }
    }

    async fn execute_run_script_task(&self, task: TaskMessage) -> ResultMessage {
        let script_content = match task.payload.get("script_content").and_then(|v| v.as_str()) {
            Some(encoded) => match general_purpose::STANDARD.decode(encoded) {
                Ok(bytes) => match String::from_utf8(bytes) {
                    Ok(script) => script,
                    Err(e) => {
                        return ResultMessage::error(
                            task.command_id,
                            self.agent_id.clone(),
                            format!("Invalid UTF-8 in script: {}", e),
                        )
                    }
                },
                Err(e) => {
                    return ResultMessage::error(
                        task.command_id,
                        self.agent_id.clone(),
                        format!("Failed to decode base64 script: {}", e),
                    )
                }
            },
            None => {
                return ResultMessage::error(
                    task.command_id,
                    self.agent_id.clone(),
                    "Missing script_content in payload".to_string(),
                )
            }
        };

        let timeout_sec = task
            .payload
            .get("timeout_sec")
            .and_then(|v| v.as_u64())
            .unwrap_or(30) as u32;

        println!(
            "🔧 Executing script (timeout: {}s): {}",
            timeout_sec,
            &script_content[..script_content.len().min(100)]
        );

        // Execute script
        match Command::new("sh").arg("-c").arg(&script_content).output() {
            Ok(output) => {
                let stdout = String::from_utf8_lossy(&output.stdout).to_string();
                let stderr = String::from_utf8_lossy(&output.stderr).to_string();
                let exit_code = output.status.code().unwrap_or(-1);

                ResultMessage::success(
                    task.command_id,
                    self.agent_id.clone(),
                    serde_json::json!({
                        "exit_code": exit_code,
                        "stdout": general_purpose::STANDARD.encode(stdout),
                        "stderr": general_purpose::STANDARD.encode(stderr)
                    }),
                )
            }
            Err(e) => ResultMessage::error(
                task.command_id,
                self.agent_id.clone(),
                format!("Failed to execute script: {}", e),
            ),
        }
    }

    async fn execute_receive_file_task(&self, task: TaskMessage) -> ResultMessage {
        let file_path = match task.payload.get("file_path").and_then(|v| v.as_str()) {
            Some(path) => path,
            None => {
                return ResultMessage::error(
                    task.command_id,
                    self.agent_id.clone(),
                    "Missing file_path in payload".to_string(),
                )
            }
        };

        let file_content = match task.payload.get("file_content").and_then(|v| v.as_str()) {
            Some(encoded) => match general_purpose::STANDARD.decode(encoded) {
                Ok(bytes) => bytes,
                Err(e) => {
                    return ResultMessage::error(
                        task.command_id,
                        self.agent_id.clone(),
                        format!("Failed to decode base64 file content: {}", e),
                    )
                }
            },
            None => {
                return ResultMessage::error(
                    task.command_id,
                    self.agent_id.clone(),
                    "Missing file_content in payload".to_string(),
                )
            }
        };

        println!(
            "📁 Receiving file: {} ({} bytes)",
            file_path,
            file_content.len()
        );

        // Write file
        match std::fs::write(file_path, &file_content) {
            Ok(_) => ResultMessage::success(
                task.command_id,
                self.agent_id.clone(),
                serde_json::json!({
                    "file_path": file_path,
                    "bytes_written": file_content.len()
                }),
            ),
            Err(e) => ResultMessage::error(
                task.command_id,
                self.agent_id.clone(),
                format!("Failed to write file {}: {}", file_path, e),
            ),
        }
    }

    async fn send_result(&self, result: ResultMessage) -> Result<()> {
        let payload = serde_json::to_vec(&result)?;
        self.rabbitmq_channel
            .basic_publish(
                queues::RESULT_EXCHANGE,
                "result",
                BasicPublishOptions::default(),
                &payload,
                BasicProperties::default(),
            )
            .await?;

        println!("📤 Result sent for command: {}", result.command_id);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();

    let groups: Vec<String> = cli
        .groups
        .split(',')
        .map(|s| s.trim().to_string())
        .collect();

    // Create TLS configuration
    let tls_config = if cli.enable_tls {
        Some(shared::tls::TlsConfig::client_only(cli.ca_path))
    } else {
        None
    };

    let agent = NodeAgent::new(&cli.rabbitmq_url, groups, cli.token, tls_config).await?;

    println!("🚀 Starting Node Agent: {}", agent.agent_id);
    println!("🏠 Hostname: {}", agent.hostname);
    println!("🔖 Groups: {:?}", agent.groups);

    agent.start().await?;

    // Keep agent running
    tokio::signal::ctrl_c().await?;
    println!("🛑 Shutting down agent...");

    Ok(())
}

#[cfg(test)]
mod tests {
    //! Unit tests for node_agent functionality

    use super::*;
    use std::process::Command as StdCommand;

    #[tokio::test]
    async fn hostname_command_execution() {
        // Test hostname command creation
        let task = TaskMessage::new_get_hostname(Some("test-group".to_string()));

        assert_eq!(task.command_type, "get_hostname");
        assert_eq!(task.target_group, Some("test-group".to_string()));

        // Test actual hostname command execution (mock)
        let output = StdCommand::new("hostname")
            .output()
            .expect("Failed to execute hostname command");

        assert!(output.status.success());
        let hostname_str = String::from_utf8_lossy(&output.stdout);
        assert!(!hostname_str.trim().is_empty());
    }

    #[tokio::test]
    async fn script_execution_task() {
        let script_content = "echo 'Hello from test'; echo 'Second line'";
        let timeout = 10;
        let task = TaskMessage::new_run_script(script_content, timeout, None);

        assert_eq!(task.command_type, "run_script");
        assert_eq!(task.payload["timeout_sec"], 10);

        // Verify script content is base64 encoded
        let encoded_script = task.payload["script_content"].as_str().unwrap();
        let decoded = base64::engine::general_purpose::STANDARD
            .decode(encoded_script)
            .unwrap();
        assert_eq!(String::from_utf8(decoded).unwrap(), script_content);
    }

    #[tokio::test]
    async fn file_receive_task() {
        let file_path = "/tmp/test-file.txt";
        let file_content = b"Test file content with binary data \x00\x01\x02";
        let task =
            TaskMessage::new_receive_file(file_path, file_content, Some("files".to_string()));

        assert_eq!(task.command_type, "receive_file");
        assert_eq!(task.target_group, Some("files".to_string()));
        assert_eq!(task.payload["file_path"], file_path);

        // Verify file content encoding
        let encoded_content = task.payload["file_content"].as_str().unwrap();
        let decoded = base64::engine::general_purpose::STANDARD
            .decode(encoded_content)
            .unwrap();
        assert_eq!(decoded, file_content);
    }

    #[test]
    fn cli_argument_parsing() {
        // Test CLI structure (would use clap::Parser in real scenario)
        let groups = "web,database,cache";
        let parsed_groups: Vec<String> = groups.split(',').map(|s| s.trim().to_string()).collect();

        assert_eq!(parsed_groups.len(), 3);
        assert_eq!(parsed_groups[0], "web");
        assert_eq!(parsed_groups[1], "database");
        assert_eq!(parsed_groups[2], "cache");
    }

    #[tokio::test]
    async fn command_timeout_scenarios() {
        // Test different timeout values
        let quick_task = TaskMessage::new_run_script("echo 'quick'", 5, None);
        let long_task = TaskMessage::new_run_script("sleep 1; echo 'done'", 60, None);

        assert_eq!(quick_task.payload["timeout_sec"], 5);
        assert_eq!(long_task.payload["timeout_sec"], 60);

        // Verify command IDs are unique
        assert_ne!(quick_task.command_id, long_task.command_id);
    }

    #[tokio::test]
    async fn agent_group_membership() {
        // Test agent with multiple groups
        let multi_group_agent = AgentRegistration::new(
            "multi-agent".to_string(),
            "multi-node".to_string(),
            vec![
                "web".to_string(),
                "database".to_string(),
                "monitoring".to_string(),
            ],
            "multi_token".to_string(),
            "Linux CentOS 8".to_string(),
        );

        assert_eq!(multi_group_agent.groups.len(), 3);
        assert!(multi_group_agent.groups.contains(&"web".to_string()));
        assert!(multi_group_agent.groups.contains(&"database".to_string()));
        assert!(multi_group_agent.groups.contains(&"monitoring".to_string()));

        // Test agent with single group
        let single_group_agent = AgentRegistration::new(
            "single-agent".to_string(),
            "single-node".to_string(),
            vec!["api".to_string()],
            "single_token".to_string(),
            "Linux Ubuntu 20.04".to_string(),
        );

        assert_eq!(single_group_agent.groups.len(), 1);
        assert!(single_group_agent.groups.contains(&"api".to_string()));
    }
}
